package com.kaoshi.myweb.snippet.mq.rabbitmq.publish;


import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * Created by Administrator on 2017/3/4.
 * 实现rpc调用,虽然，跟rabbitmq耦合了，以后，需要前端js调用就懵逼了，特定场景可用
 * consumer-->RPCClient
 */
public class RPCServer {

	private static final String RPC_QUEUE_NAME = "rpc_queue";

	public static void main(String[] argv) throws IOException, TimeoutException {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");

		Connection connection = null;
		try {
			connection      = factory.newConnection();
			final Channel channel = connection.createChannel();

			channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);

			channel.basicQos(1);

			System.out.println(" [x] Awaiting RPC requests");

			Consumer consumer = new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
					AMQP.BasicProperties replyProps = new AMQP.BasicProperties
							.Builder()
							.correlationId(properties.getCorrelationId())
							.build();

					String response = "";

					try {
						String message = new String(body,"UTF-8");
						int n = Integer.parseInt(message);

						System.out.println(" [.] fib(" + message + ")");
						response += fib(n);
					}
					catch (RuntimeException e){
						System.out.println(" [.] " + e.toString());
					}
					finally {
						channel.basicPublish( "", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));

						channel.basicAck(envelope.getDeliveryTag(), false);
					}
				}
			};

			channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

			//...
		}finally {

		}
	}

	private static int fib(int n) {
		if (n == 0) return 0;
		if (n == 1) return 1;
		return fib(n-1) + fib(n-2);
	}
}